﻿#include "ArkXbpClient.h"
#include "ArkXbpManag.h"

namespace kd
{
	ArkXbpClient::ArkXbpClient(const ArkKcxpSvr& xpSvrinfo)
	{
		mXpServer = xpSvrinfo;
		mkGroupID = ArkXbpManag::GetGroupID();

		mkMsgLifeTime = -1;
		mkRecvBuffsize = xpSvrinfo.kcxp_buffsize;

		mkRecvBuffer = new char[mkRecvBuffsize];
		memset(mkRecvBuffer, 0, sizeof(mkRecvBuffer));

		mkBrowseMode = 0;
		mkTimeStamp = { 0 };

		kcxp_reconnect_callback_ = nullptr;
		kcxp_trace_callback_ = nullptr;
		callback_user_data_ = nullptr;

		mkNeedReconnect = false;
		mkDisconByNodata = false;

		std::ostringstream ossOption;
		ossOption << xpSvrinfo.kcxp_addr << ":" << xpSvrinfo.kcxp_port;
		mkXpOption = ossOption.str();
	}

	ArkXbpClient::~ArkXbpClient()
	{
		Disconnect();

		if (mkRecvBuffer)
		{
			delete[]mkRecvBuffer;
			mkRecvBuffer = nullptr;
		}
	}

	time_t ArkXbpClient::GetTimeStamp()
	{
		return mkTimeStamp;
	}

	void ArkXbpClient::SetTimeStamp(time_t when)
	{
		mkTimeStamp = when;
	}

	int ArkXbpClient::GetConTimeOut()
	{
		return mXpServer.kcxp_timeout;
	}

	void ArkXbpClient::SetConTimeOut(int secs)
	{
		mXpServer.kcxp_timeout = secs;
	}

	int ArkXbpClient::GetMaxWaitTimes()
	{
		return mkMaxWaitTimes;
	}

	void ArkXbpClient::SetMaxWaitTimes(int times)
	{
		mkMaxWaitTimes = times;
	}

	std::string ArkXbpClient::GetGroupID()
	{
		return mkGroupID;
	}

	void ArkXbpClient::SetGroupID(std::string groupid)
	{
		mkGroupID = groupid;
	}

	std::string ArkXbpClient::GetLastErrorInfo()
	{
		return mkErrorMsg;
	}

	std::string ArkXbpClient::GetKcxpDesc()
	{
		return mkXpOption;
	}

	std::string ArkXbpClient::GetQueueName()
	{
		return mXpServer.kcxp_reqque;
	}

	int ArkXbpClient::Connect()
	{
		int retCode = POWERPCB_OK;
		if ((retCode = OpenQueueMgr()) != POWERPCB_OK ||
			(retCode = OpenQueueNsv()) != POWERPCB_OK)
			return retCode;

		mkNeedReconnect = false;

		return POWERPCB_OK;
	}

	int ArkXbpClient::Disconnect()
	{
		try
		{
			int retCode = POWERPCB_OK;
			if ((retCode = CloseQueueNsv()) != POWERPCB_OK ||
				(retCode = CloseQueueMgr()) != POWERPCB_OK)
			{
				return retCode;
			}
		}
		catch (...)
		{
			// TODO: 这里断开连接的时候会抛异常，暂时没有去深究
			ARK_LOG_ERROR("An unkown error occurred while disconnect to kcxp");
		}

		mkNeedReconnect = true;

		return POWERPCB_OK;
	}

	int ArkXbpClient::Get(std::string &msgid, std::string &data, int correlid)
	{
		int retCode = POWERPCB_OK;
		if ((retCode = Get(msgid, correlid)) != POWERPCB_OK)
			return retCode;

		data = mkRecvBuffer;

		return POWERPCB_OK;
	}

	int ArkXbpClient::Get(std::string &msgid, char*data, int len, int correlid)
	{
		if (!data || len < 1)
		{
			mkErrorMsg = "invalid recv buffer";
			return POWERPCB_ERR;
		}

		// 检查是否需要重新连接
		if (POWERPCB_OK != CanReconnect())
		{
			mkErrorMsg = "can not open kcxp queue";
			return POWERPCB_ERR;
		}

		// 检查是否已经连接KCXP
		if (mkHconn == kInvalidHandle || mkHobj == kInvalidHandle)
		{
			mkErrorMsg = "can not connect to kcxp server";
			return POWERPCB_ERR;
		}

		KCXP_LONG datalen;
		KCXP_LONG compcode, reason;
		// 构建本次KCXP_Get选项和消息描述
		InitGetMsgOptsAndDesc(msgid, correlid);

		data[0] = 0;
		XpBpHelper::KCXP_Get(mkHconn, mkHobj, &mkGetMsgDesc, &mkGetMsgOpts, len, data, &datalen, &compcode, &reason);
		if (compcode != KCXP_CC_OK)
		{
			// 没有满足条件的消息，一般是超时导致
			if (reason == KCXP_RC_NO_MSG_AVAILABLE)
			{
				if (mkBrowseMode)
				{
					mkErrorMsg = "receive msg from kcxp time out";
					return reason;
				}
			}

			// 检查KCXP连接有效性
			if (IsBrokenConn(reason)) mkNeedReconnect = true;
			// 格式化错误信息描述
			mkErrorMsg = StringUtilFormat("code：%d message：%s", reason, GetKcxpErrInfo(reason).c_str());

			return reason;
		}

		// 结束符号
		data[datalen] = 0;
		// 返回消息标识
		if (msgid.empty()) msgid = std::string(mkGetMsgDesc.strMsgId, sizeof(mkGetMsgDesc.strMsgId));

		return POWERPCB_OK;
	}

	int ArkXbpClient::Put(std::string &msgid, const std::string &data, int correlid)
	{
		// 检查是否需要重新连接
		if (POWERPCB_OK != CanReconnect()) {
			mkErrorMsg = "can not open kcxp queue";
			return POWERPCB_ERR;
		}
		// 检查是否已经连接KCXP 
		if (mkHconn == kInvalidHandle || mkHobj == kInvalidHandle) {
			mkErrorMsg = "can not connect to kcxp server";
			return POWERPCB_ERR;
		}

		// 构建本次KCXP_Put选项和消息描述
		InitPutMsgOptsAndDesc(msgid, data.size(), correlid);

		// 将报文写入KCXP
		KCXP_LONG compcode, reason;
		XpBpHelper::KCXP_Put(mkHconn, mkHobj, &mkPutMsgDesc, &mkPutMsgOpts, (KCXP_LONG)data.size(), (void *)data.c_str(), &compcode, &reason);
		if (compcode == KCXP_CC_FAILED)
		{
			if (IsBrokenConn(reason)) mkNeedReconnect = true;
			mkErrorMsg = StringUtilFormat("can not write kcxp queue，code：%d，message：[%ld]%s", compcode, reason, GetKcxpErrInfo(reason).c_str());
		}

		// 返回报文消息标识
		if (msgid.empty()) msgid = std::string(mkPutMsgDesc.strMsgId, sizeof(mkPutMsgDesc.strMsgId));

		return reason;
	}

	int ArkXbpClient::Put(std::string &msgid, const char*data, int len, int correlid)
	{
		// 检查是否需要重新连接
		if (POWERPCB_OK != CanReconnect()) {
			mkErrorMsg = "can not open kcxp queue";
			return POWERPCB_ERR;
		}
		// 检查是否已经连接KCXP 
		if (mkHconn == kInvalidHandle || mkHobj == kInvalidHandle) {
			mkErrorMsg = "can not connect to kcxp server";
			return POWERPCB_ERR;
		}

		// 构建本次KCXP_Put选项和消息描述
		InitPutMsgOptsAndDesc(msgid, len, correlid);

		// 将报文写入KCXP
		KCXP_LONG compcode, reason;
		XpBpHelper::KCXP_Put(mkHconn, mkHobj, &mkPutMsgDesc, &mkPutMsgOpts, (KCXP_LONG)len, (void *)data, &compcode, &reason);
		if (compcode == KCXP_CC_FAILED)
		{
			if (IsBrokenConn(reason)) mkNeedReconnect = true;
			mkErrorMsg = StringUtilFormat("can not write kcxp queue，code：%d，message：[%ld]%s", compcode, reason, GetKcxpErrInfo(reason).c_str());
		}

		// 返回报文消息标识
		if (msgid.empty()) msgid = std::string(mkPutMsgDesc.strMsgId, sizeof(mkPutMsgDesc.strMsgId));

		return reason;
	}

	std::string ArkXbpClient::StringUtilFormat(const char *pMsg, ...)
	{
		va_list ap;
		char text[KCBP_LOG_BUFFER_LEN + 1];
		va_start(ap, pMsg);
		vsnprintf(text, KCBP_LOG_BUFFER_LEN, pMsg, ap);
		va_end(ap);

		return std::string(text);
	}

	std::string ArkXbpClient::GetKcxpErrInfo(int reason)
	{
		char english_error_info[8192];
		char chinese_error_info[8192];
		english_error_info[0] = '\0';
		chinese_error_info[0] = '\0';
		XpBpHelper::KCXP_ErrInfo(reason, english_error_info, chinese_error_info);

		return std::string(english_error_info);
	}

	bool ArkXbpClient::IsBrokenConn(int reason)
	{
		if (reason == KCXP_RC_CONNECTION_BROKEN ||
			reason == KCXP_RC_CONNECTION_STOPPING ||
			reason == KCXP_RC_HCONN_ERROR ||
			reason == KCXP_RC_HOBJ_ERROR/* ||*/
			/*reason == KCXP_RC_DATA_ERROR*/) {
			return true;
		}

		return false;
	}

	int ArkXbpClient::Get(std::string &msgid, int correlid)
	{
		// 检查是否需要重新连接
		if (POWERPCB_OK != CanReconnect())
		{
			mkErrorMsg = "can not open kcxp queue";
			return POWERPCB_ERR;
		}

		// 检查是否已经连接KCXP
		if (mkHconn == kInvalidHandle || mkHobj == kInvalidHandle)
		{
			mkErrorMsg = "can not connect to kcxp server";
			return POWERPCB_ERR;
		}

		KCXP_LONG datalen;
		KCXP_LONG compcode, reason;
		KCXP_INT bufflen = mkRecvBuffsize;
		// 构建本次KCXP_Get选项和消息描述
		InitGetMsgOptsAndDesc(msgid, correlid);

		mkRecvBuffer[0] = 0;
		XpBpHelper::KCXP_Get(mkHconn, mkHobj, &mkGetMsgDesc, &mkGetMsgOpts, bufflen, mkRecvBuffer, &datalen, &compcode, &reason);
		if (compcode != KCXP_CC_OK)
		{
			// 没有满足条件的消息，一般是超时导致
			if (reason == KCXP_RC_NO_MSG_AVAILABLE)
			{
				if (mkBrowseMode)
				{
					mkErrorMsg = "receive msg from kcxp time out";
					return reason;
				}
			}

			// 检查KCXP连接有效性
			if (IsBrokenConn(reason)) mkNeedReconnect = true;
			// 格式化错误信息描述
			mkErrorMsg = StringUtilFormat("code：%d message：%s", reason, GetKcxpErrInfo(reason).c_str());

			return reason;
		}

		// 结束符号
		mkRecvBuffer[datalen] = 0;
		// 返回消息标识
		if (msgid.empty()) msgid = std::string(mkGetMsgDesc.strMsgId, sizeof(mkGetMsgDesc.strMsgId));

		return POWERPCB_OK;
	}

	std::string ArkXbpClient::GetRecvData()
	{
		return mkRecvBuffer;
	}

	int ArkXbpClient::OpenQueueMgr()
	{
		KCXP_LONG  compcode, reason;

		try
		{
			// 直连KCXP队列管理器
			XpBpHelper::KCXP_Connx((char*)mXpServer.kcxp_addr.c_str(), mXpServer.kcxp_port, (char*)mXpServer.kcxp_user.c_str(), (char*)mXpServer.kcxp_pswd.c_str(), &mkHconn, &compcode, &reason);
			if (compcode != KCXP_CC_OK)
			{
				mkErrorMsg = StringUtilFormat("KCXP<%s> connect failed，message：[%ld]%s", mkXpOption.c_str(), reason, GetKcxpErrInfo(reason).c_str());
			}
		}
		catch (...)
		{
			mkErrorMsg = "打开队列管理器时发生异常!";
		}

		return reason;
	}

	int ArkXbpClient::CloseQueueMgr()
	{
		KCXP_LONG  compcode, reason = POWERPCB_OK;

		try
		{
			if (mkHconn != kInvalidHandle)
			{
				XpBpHelper::KCXP_Disconn(&mkHconn, 0L, &compcode, &reason);
				if (compcode != KCXP_CC_OK)
				{
					mkErrorMsg = StringUtilFormat("KCXP close failed，code：%d：message：[%ld]%s", compcode, reason, GetKcxpErrInfo(reason).c_str());
					return reason;
				}
				mkHconn = kInvalidHandle;
				mkHobj = kInvalidHandle;
			}
		}
		catch (...)
		{
			mkErrorMsg = "An exception occurred while closing the queue manager!";
		}

		return reason;
	}

	int ArkXbpClient::OpenQueueNsv()
	{
		KCXP_OD objdesc = { KCXP_DEFAULT_OD };
		objdesc.iObjectType = KCXP_OT_Q;

		strncpy(objdesc.strObjectName, mXpServer.kcxp_reqque.c_str(), sizeof(objdesc.strObjectName) - 1);
		int open_option = KCXP_OO_AS_Q_DEF | KCXP_OO_OUTPUT | KCXP_OO_INPUT | KCXP_OO_INQUIRE;

		KCXP_LONG compcode, reason;
		XpBpHelper::KCXP_Open(mkHconn, &objdesc, open_option, &mkHobj, &compcode, &reason);
		if (compcode != KCXP_CC_OK)
		{
			mkErrorMsg = StringUtilFormat("KCXP<%s> can not open queue，message：[%ld]%s", mXpServer.kcxp_reqque.c_str(), reason, GetKcxpErrInfo(reason).c_str());
		}

		return reason;
	}

	int ArkXbpClient::CloseQueueNsv()
	{
		KCXP_LONG  compcode;
		KCXP_LONG  reason = POWERPCB_OK;

		if (mkHconn != kInvalidHandle && mkHobj != kInvalidHandle)
		{
			XpBpHelper::KCXP_Close(mkHconn, &mkHobj, KCXP_CO_NONE, &compcode, &reason);
			if (compcode != KCXP_CC_OK)
			{
				mkErrorMsg = StringUtilFormat("KCXP<%s>can not open queue，message：[%ld]%s", mXpServer.kcxp_reqque.c_str(), reason, GetKcxpErrInfo(reason).c_str());
			}
		}

		return reason;
	}

	int ArkXbpClient::JustReconnect()
	{
		int retCode = Disconnect();

		if (retCode == POWERPCB_OK)
			retCode = Connect();

		return retCode;
	}

	int ArkXbpClient::CanReconnect()
	{
		int retCode = POWERPCB_OK;
		if (mkNeedReconnect)
		{
			if ((retCode = JustReconnect()) != POWERPCB_OK)
			{
				mkErrorMsg = "reconnect to kcxp server failed";
				usleep(1000);
			}

			if (kcxp_reconnect_callback_)
			{
				// 启用重连回调函数
				kcxp_reconnect_callback_(callback_user_data_, retCode);
			}
		}

		return retCode;
	}

	long ArkXbpClient::GetQueueDepth()
	{
		KCXP_LONG  compcode, reason = POWERPCB_OK;
		KCXP_LONG  selector_count, selectors[1];
		KCXP_LONG  attr_count, attrs[1];

		selector_count = 1;
		selectors[0] = KCXP_QA_CURDEPTH;

		attr_count = 1;

		XpBpHelper::KCXP_Inq(mkHconn, mkHobj, selector_count, selectors, &attr_count, attrs, NULL, NULL, &compcode, &reason);

		long depth = -1;
		if (compcode != KCXP_CC_OK)
		{
			mkErrorMsg = StringUtilFormat("Failed to query kcxp queue depth，message：[%ld]%s", reason, GetKcxpErrInfo(reason).c_str());
		}
		else
		{
			depth = attrs[0];
		}

		return depth;
	}

	void ArkXbpClient::InitGetMsgOptsAndDesc(const std::string &strMsgId, int correlId)
	{
		mkGetMsgOpts = { KCXP_DEFAULT_GMO };
		mkGetMsgDesc = { KCXP_DEFAULT_MD };

		// KCXP_GMO_WAIT 等待方式读取消息
		// KCXP_GMO_FAIL_IF_QUIESCING 判断队列管理器是否正常状态
		mkGetMsgOpts.Options |= KCXP_GMO_WAIT | KCXP_GMO_FAIL_IF_QUIESCING | KCXP_GMO_GET;
		mkGetMsgOpts.WaitInterval = mXpServer.kcxp_timeout;

		if (!mkGroupID.empty())
		{
			//共享队列时，根据GOUPID取值
			strncpy(mkGetMsgDesc.strGroupId, mkGroupID.c_str(), sizeof(mkGetMsgDesc.strGroupId));
			strncpy(mkGetMsgDesc.strCorrelId, std::to_string(correlId).c_str(), sizeof(mkGetMsgDesc.strCorrelId));

			mkGetMsgOpts.MatchOptions |= KCXP_MO_MATCH_GROUP_ID | KCXP_MO_MATCH_CORREL_ID;
		}

		if (!strMsgId.empty())
		{
			//非共享队列时，指定MSGID取值
			strncpy(mkGetMsgDesc.strMsgId, strMsgId.c_str(), sizeof(mkGetMsgDesc.strMsgId));

			//匹配MSGID方式取值
			mkGetMsgOpts.MatchOptions |= KCXP_MO_MATCH_MSG_ID;
			if (mkBrowseMode) {
				mkGetMsgOpts.Options |= KCXP_GMO_NO_WAIT | KCXP_GMO_BROWSE_FIRST;
			}
		}
	}

	void ArkXbpClient::InitPutMsgOptsAndDesc(const std::string &msgid, size_t datalen, int correlid)
	{
		mkPutMsgOpts = { KCXP_DEFAULT_PMO };
		mkPutMsgDesc = { KCXP_DEFAULT_MD };

		//缺省10秒超时
		mkPutMsgOpts.Timeout = mXpServer.kcxp_timeout;
		if (mkPutMsgOpts.Timeout <= 0) mkPutMsgOpts.Timeout = 10;

		// 消息超过32k分段
		mkPutMsgDesc.cbMsgFlags |= KCXP_MF_SEGMENTATION_ALLOWED;
		mkPutMsgDesc.iPriority = KCXP_PRIORITY_5;
		mkPutMsgDesc.cbPersistence = KCXP_NOPERSISTENCE_MESSAGE;
		mkPutMsgDesc.cbMsgType = KCXP_MT_DATAGRAM;
		mkPutMsgDesc.iLifeTime = mkMsgLifeTime;
		mkPutMsgDesc.iDataLength = (KCXP_LONG)datalen;

		// 设置消息生命周期
		if (mkPutMsgDesc.iLifeTime <= 0)
			mkPutMsgDesc.iLifeTime = 12 * 60 * 60 * 1000;

		// 设置应答队列名称
		if (!mXpServer.kcxp_ansque.empty() && mXpServer.kcxp_reqque != mXpServer.kcxp_ansque)
			strncpy(mkPutMsgDesc.strReplyToQ, mXpServer.kcxp_ansque.c_str(), sizeof(KCXP_CHAR32));

		// 设置消息集群标识
		if (!mkGroupID.empty())
			strncpy(mkPutMsgDesc.strGroupId, mkGroupID.c_str(), sizeof(KCXP_CHAR24));

		// 设置消息标识
		// 假设msgid长度刚好24字节，strncpy拷贝字符串没有结束符号，strncpy_s有结束符，拷贝长度为23字节
		if (!msgid.empty())
			strncpy(mkPutMsgDesc.strMsgId, msgid.c_str(), sizeof(mkPutMsgDesc.strMsgId));
		//strncpy_s(mkPutMsgDesc.strMsgId, sizeof(KCXP_CHAR24), msgid.c_str(), _TRUNCATE);
		else
			mkPutMsgOpts.Options = KCXP_PMO_NEW_MSG_ID;

		// 设置目标节点
		if (!mkDstNode.empty())
			strncpy(mkPutMsgDesc.strDestNodeCode, mkDstNode.c_str(), sizeof(KCXP_CHAR9));

		// 读取报文编号
		//std::string serialNum = data.substr(PACKETINDEX_SITE, PACKETINDEX_SIZE);

		// 注意KCBP只能取 correid为"0"的请求
		strncpy(mkPutMsgDesc.strCorrelId, std::to_string(correlid).c_str(), sizeof(KCXP_CHAR24));
	}

	ArkKcbpClient::ArkKcbpClient(const ArkKcxpSvr& xpSvrinfo)
		: ArkXbpClient(xpSvrinfo)
	{

	}

	int ArkKcbpClient::Get(std::string &msgid, std::string &data, int correlid)
	{
		int waitTmoutTimes = 0;
		bool recvContinue = false;
		if (!data.empty())
			data.clear();

		do
		{
			std::string recvData;
			int retCode = POWERPCB_OK;

			// 本次读取计时器
			if ((retCode = ArkXbpClient::Get(msgid, recvData, correlid)) != POWERPCB_OK)
			{
				// 检查"没有满足条件的消息"错误代码，一般超时导致
				if (retCode == KCXP_RC_NO_MSG_AVAILABLE)
				{
					// 继续接收消息
					recvContinue = true;

					// 超时次数太多，配置重新链接KCXP
					if (0 < mkMaxWaitTimes && 0 == (++waitTmoutTimes % (mkMaxWaitTimes + 1)))
					{
						mkNeedReconnect = true;
						return retCode;
					}

					if (!msgid.empty())
					{
						mkNeedReconnect = true;
						// 多组分包中某个分包接收超时，尝试重连
						mkErrorMsg = StringUtilFormat("[%d]read message[%s]time out[%d][%s] queue[%s]thread[%d]", correlid, msgid.c_str(),
							waitTmoutTimes, mkErrorMsg.c_str(), mXpServer.kcxp_ansque.c_str(), kd::gettid());
						// 打印读取KCXP消息失败细节，更新次数
						ARK_LOG_ERROR("%s", mkErrorMsg.c_str());
						// 若没有配置接收报文失败重试，直接返回
						if (mkMaxWaitTimes < 1)
							return retCode;
					}

					if (mkNeedReconnect)
					{
						// 设置重新获取数据correlid = 0 msgid.clear() data.clear();
						if (kcxp_trace_callback_) kcxp_trace_callback_(callback_user_data_, 0, mkErrorMsg.c_str());
					}
					continue;
				}

				return retCode;
			}

			if (waitTmoutTimes) ARK_LOG_INFO(">[%d]read message [%s] success [%d][code：0000 message：successfully read message after timeout ] queue[%s]thread[%d]",
				correlid, msgid.c_str(), waitTmoutTimes, mXpServer.kcxp_ansque.c_str(), kd::gettid());

			// 接收报文长度校验
			size_t headCooklen = KCBPPACKETHEADRAW_SIZE;
			if (KCBPPACKETHEADRAW_SIZE > recvData.length())
			{
				ARK_LOG_ERROR("system detects abnormal kcbp message，length：%u", recvData.length());
				return POWERPCB_ERR;
			}

			// 解析KCBP报文头部结构
			KCBPPacketHeadCooked kcbp_head;
			KCBPPacket *kcbpPacket = (KCBPPacket *)recvData.c_str();
			// 解码KCBP报文头部结构
			PacketHeadRaw2Cooked(kcbp_head, kcbpPacket->stKCBPPacketHeadRaw);

			// 接收报文校验是否正确
			if (0 && kcbp_head.Parity != 0 && VerifyParity(*kcbpPacket) != 0)
			{
				ARK_LOG_ERROR("system detects abnormal packet verification，correlid：%d", correlid);
				return POWERPCB_ERR;
			}

			// 接收报文检查消息序号
			if (kcbp_head.PacketIndex != correlid)
			{
				ARK_LOG_ERROR("system detects that the serial number of the received message is abnormal，wish：%d，reality：%d", kcbp_head.PacketIndex, correlid);
				return POWERPCB_ERR;
			}

			// 首个报文增加头部结构
			if (correlid == 0) data += std::string(mkRecvBuffer, KCBPPACKETHEADRAW_SIZE);
			// 报文添加数据内容信息
			data += std::string((char *)kcbpPacket->PacketBody, kcbp_head.PacketLength);
			// 检查是否继续接收数据
			if ((recvContinue = (!(kcbp_head.EOP & 1))))
				++correlid;

		} while (recvContinue);

		return POWERPCB_OK;
	}

	int ArkKcbpClient::Put(std::string &msgid, const std::string &data, int correlid)
	{
		// 小于4K报文无需分包发送
		if (data.size() <= KCBP_SEND_MAX_BUF_SIZE)
			return ArkXbpClient::Put(msgid, data, correlid);

		// 构建Put选项和消息描述
		InitPutMsgOptsAndDesc(msgid, data.size(), correlid);

		return Write(msgid, data, correlid);
	}

	int ArkKcbpClient::Write(std::string &msgid, const std::string &data, int correlid)
	{
		// 报文索引
		int pktIndex = correlid;
		// 报文长度
		int dataLength = (int)data.size();
		// 解析报文头部结构
		struct KCBPPacket *telegramProfile = (struct KCBPPacket *)data.c_str();

		KCBPPacketHeadCooked kcbpHeader;
		PacketHeadRaw2Cooked(kcbpHeader, telegramProfile->stKCBPPacketHeadRaw);
		// 报文发送偏移位置
		int dataSendoffset = sizeof(KCBPPacketHeadRaw);
		// 报文内容极限长度
		int kcbpBodyMaxSize = KCBP_SEND_MAX_BUF_SIZE - sizeof(struct KCBPPacketHeadRaw);
		// 报文内容消息标识
		memset(mkPutMsgDesc.strCorrelId, 0, sizeof(mkPutMsgDesc.strCorrelId));

		int sendPktLength;
		do
		{
			kcbpHeader.EOP = 0;
			snprintf(mkPutMsgDesc.strCorrelId, sizeof(KCXP_CHAR24), "%d", pktIndex);
			kcbpHeader.PacketIndex = pktIndex++;

			if (dataLength - dataSendoffset > kcbpBodyMaxSize)
			{
				sendPktLength = kcbpBodyMaxSize;
				kcbpHeader.EOP |= 0;			//中间packet
			}
			else
			{
				sendPktLength = dataLength - dataSendoffset;
				kcbpHeader.EOP |= ENDOFPACKAGE; //最后packet
			}

			kcbpHeader.PacketLength = sendPktLength;

			PacketHeadCooked2Raw(kcbp.stKCBPPacketHeadRaw, kcbpHeader);

			strncpy((char *)kcbp.PacketBody, (char *)telegramProfile + dataSendoffset, sendPktLength);
			SetParity(kcbp);

			int retCode = ArkXbpClient::Put(msgid, std::string((char *)&kcbp, sizeof(KCBPPacketHeadRaw) + sendPktLength), kcbpHeader.PacketIndex);
			if (retCode != POWERPCB_OK)
			{
				mkErrorMsg = StringUtilFormat("write kcxp queue failed，%s", GetKcxpErrInfo(retCode).c_str());
				return retCode;
			}

			dataSendoffset += sendPktLength;
		} while (dataSendoffset < dataLength);

		return POWERPCB_OK;
	}

}